"""
综合案例：搜索引擎日志分析
需求：
1、读取文件转成RDD
2、打印输出：热门搜索时间段（小时精度）TOP3
3、打印输出：热门搜索词TOP3
4、打印输出：统计黑马程序员关键字在哪个时间段被搜索的最多
5、将数据转换为JSON格式，写入文件
"""
from pyspark import SparkContext, SparkConf
import os

os.environ['PYSPARK_PYTHON'] = "D:/Software/Work/miniconda3/python.exe"
os.environ['HADOOP_HOME'] = "D:/Software/Work/hadoop-3.0.0"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
conf.set("spark.default.parallelism", "1")
sc = SparkContext(conf=conf)

# TODO 需求1：读取文件转成RDD
rdd = sc.textFile("search_log.txt")
# 将每行数据使用\t分割，使每行元素变成一个list
data_rdd = rdd.map(lambda x: x.split("\t"))

# TODO 需求2：打印输出：热门搜索时间段（小时精度）TOP3
# 1、取出全部的时间并只保留小时部分（可以用split方法，也可以用切片）
# 2、转换为(小时, 1)的二元元组（map算子）
# 3、按key分组聚合value（累加）（reduceByKey算子）
# 4、根据聚合后的次数（降序）从大到小排序（sortBy算子）
# 5、取前3个结果（take算子）
hot_time_top_3 = data_rdd.map(lambda x: (x[0][:2], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False, numPartitions=1) \
    .take(3)
print(f"热门搜索时间段TOP3：{hot_time_top_3}")

# TODO 需求3：打印输出：热门搜索词TOP3
# 1、取出每个搜索记录的搜索词
# 2、转换为（搜索词， 1）的二元元组（map算子）
# 3、按key分组聚合value（累加）（reduceByKey算子）
# 4、根据聚合后的次数（降序）从大到小排序（sortBy算子）
# 5、取前3个结果（take算子）
hot_keyword_top_3 = data_rdd.map(lambda x: (x[2], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False, numPartitions=1) \
    .take(3)
print(f"热门搜索词TOP3：{hot_keyword_top_3}")

# TODO 需求4：打印输出：统计黑马程序员关键字在哪个时间段被搜索的最多
# 1、将所有搜索词是黑马程序员的记录过滤出来（filter算子）
# 2、取出对应搜索记录的小时并组成（小时， 1）的二元元组（map算子）
# 3、按key分组聚合value（累加）（reduceByKey算子）
# 4、根据聚合后的次数（降序）从大到小排序（sortBy算子）
# 5、取第1个结果（take算子）
heima_top_time = data_rdd.filter(lambda x: x[2] == "黑马程序员") \
    .map(lambda x: (x[0].split(":")[0], 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .sortBy(lambda x: x[1], ascending=False, numPartitions=1) \
    .take(1)
print(f"“黑马程序员”关键字在{heima_top_time[0][0]}点时段被搜索的最多，次数为{heima_top_time[0][1]}")

# TODO 需求5：将数据转换为JSON格式，写入文件
# 1、转换为JSON格式的RDD
# 2、写出为文件
data_rdd.map(lambda x: {"time": x[0], "user_id": x[1], "key_word": x[2], "rank1": x[3], "rank2": x[4], "url": x[5]})\
    .saveAsTextFile("output_json")

sc.stop()
